package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

import scala.collection.mutable

object SparkStreaming03_Source_Queue {
    //队列流
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamWordCount")

    //构建流式数据的上下文环境对象，并且设置处理周期
    val ssc = new StreamingContext(conf,Seconds(3))

    val rddQueue = new mutable.Queue[RDD[Int]]
    val que: InputDStream[Int] = ssc.queueStream(rddQueue)

    que.print()



    //启动数据采集器,采集不可停止
    ssc.start()

    for (i <- 1 to 5) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }


    //主线程需要阻塞,否则main完成则释放。
    ssc.awaitTermination()






    // ssc.stop()

  }
}
